package main.scala.demo

import org.apache.spark.{Partitioner, SparkConf, SparkContext}

import scala.collection.mutable

/**
  * ActiveWebPartitionRDD
  *
  * @author zhangyimin
  *  2018-10-08 下午4:18
  * @version 1.0
  */
object ActiveWebPartitionRDD {

  def main(args: Array[String]): Unit = {

    //windows上要配置的一段代码
//    System.setProperties("hadoop.home.dir","D:\\temp\\hadoop2.7.3")

    //配置
    val conf = new SparkConf()
    conf.setAppName("webCount")
    conf.setMaster("local")
    //创建一个sparkContext
    val sc = new SparkContext(conf)


    val rdd1 = sc.textFile("hdfs://10.16.7.36:9000/data/input/sogou/localhost_access_log.2017-07-30.txt")

    //    rdd1.map(x=>(x.substring(x.indexOf("\"")+1,x.lastIndexOf("\"")).split(" ")(1),1))
    //      .reduceByKey(_+_)
    //      .saveAsTextFile("hdfs://10.16.7.36:9000/data/output/spark/100801")
    //192.168.88.1 - - [30/Jul/2017:12:54:42 +0800] "GET /MyDemoWeb/oracle.jsp HTTP/1.1" 200 242
    val rdd2 = rdd1.map {
      line => {
        val index1 = line.indexOf("\"") + 1
        val index2 = line.lastIndexOf("\"")
        val content = line.substring(index1, index2)

        val contents = content.split(" ")
        val jspContent = contents(1)

        val jspIndex = jspContent.lastIndexOf("/")
        val jspPage = jspContent.substring(jspIndex + 1)


        (jspPage, line)
      }
    }

    //根据JSP的名字建立分区,得到jsp名字的个数
    //得到所有不重复的JSP的名字
    val rdd3=rdd2.map(_._1).distinct.collect()
    //创建分区规则
    val myPartitioner=new MyPartitioner(rdd3)
    //rdd2是一个PairRDDFunctions
    //执行分区
    val res=rdd2.partitionBy(myPartitioner)
//    println(rdd3.toBuffer)
    res.saveAsTextFile("hdfs://10.16.7.36:9000/data/output/spark/100805")
    sc.stop()


  }
}

/**
  * 实现抽象类partitioner,自定义分区
  * @param allJspName
  */
class MyPartitioner(allJspName:Array[String]) extends  Partitioner{


  //定义一个集合保存分区条件
  //string jsp的名字
  //Int  分区号
  val partitionerMap=new mutable.HashMap[String,Int]()

  //建立分区规则
  var partid=0
  for(name<-allJspName){
    partitionerMap.put(name,partid)
    partid+=1
  }

  //返回分区个数
  override def numPartitions: Int = partitionerMap.size

  //根据JSP的名字获取分区号
  //KEY 就是JSP的名字
  override def getPartition(key: Any): Int = {
    partitionerMap.getOrElse(key.toString,0)
  }

}
